package org.sample.mqtt.endpoint;

import jakarta.annotation.Resource;
import org.sample.mqtt.component.JsonUtils;
import org.sample.mqtt.component.model.Connected;
import org.sample.mqtt.component.model.Disconnected;
import org.sample.mqtt.component.model.MqttResponse;
import org.sample.mqtt.service.MessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

/**
 * Created by Alan on 2020/7/18.
 */
@Component
public class MqttConsumer implements MessageHandler {

    private static final Logger log = LoggerFactory.getLogger(MqttConsumer.class);

    @Resource
    private MessageService messageService;

    @ServiceActivator(inputChannel = "mqttInboundChannel")
    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        String topic = message.getHeaders().get(MqttHeaders.TOPIC, String.class);
        Object payload = message.getPayload();
        log.info("{}\t{}", topic, payload);

        if (topic.startsWith("$")) {
            //TODO emq支持设备上线&离线通知
            if (topic.endsWith("disconnected")) {
                JsonUtils.toObj((String) payload, Disconnected.class);
            } else if (topic.endsWith("connected")) {
                JsonUtils.toObj((String) payload, Connected.class);
            }
        } else {
            try {
                //需要响应的同步类消息
                if (payload instanceof MqttResponse) {
                    messageService.response((Message<MqttResponse>) message);
                } else {
                    //TODO 服务端业务处理
                }
            } catch (Exception e) {
                log.error("\n系统出错: " + JsonUtils.toJson(payload), e);
            }
        }
    }
}